package iwb.utils.kafkapatition;

import araf.jobexecution.vo.ArafPubSubKey;
import com.google.common.collect.Lists;
import iwb.utils.RedisUtils;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.aop.aspectj.MethodInvocationProceedingJoinPoint;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.lang.reflect.Method;
import java.util.List;


@Component
@Aspect
public class PartitionExecAspect {

    public static final int PARTITION_SIZE = 1000;
    @Autowired
    PartitionExecPublish partitionExecPublish;

    @Autowired
    RedisUtils redisUtils;


    @Pointcut("@annotation(iwb.utils.kafkapatition.PartitionExecSend)")
    private void matchAbsCache() {
        throw new UnsupportedOperationException();
    }


    /**
     * around.
     *
     * @param joinPoint ProceedingJoinPoint
     * @return Object
     */
    @Around("matchAbsCache()")
    public Object around(ProceedingJoinPoint joinPoint) {

        MethodInvocationProceedingJoinPoint methodInvocationProceedingJoinPoint = (MethodInvocationProceedingJoinPoint) joinPoint;
        Method pointMethod = ((MethodSignature) methodInvocationProceedingJoinPoint.getSignature()).getMethod();
        PartitionExecSend partitionExecSend = pointMethod.getAnnotation(PartitionExecSend.class);
        redisUtils.setRedisKey(partitionExecSend.topic());
        Object[] args = joinPoint.getArgs();
        List data = (List) args[0];
        List<List> lists = Lists.partition(data, PARTITION_SIZE);
        final int[] i = {0};
        lists.forEach(list -> {
            String index = i[0] * PARTITION_SIZE + "-" + (++i[0] * PARTITION_SIZE);
            partitionExecPublish.setDataTypeId(partitionExecSend.topic());
            Messager messager = new Messager();
            messager.addData(list);
            messager.setIndexScope(index);
            partitionExecPublish.send(ArafPubSubKey.of(index), messager);
            redisUtils.setRedisKeyValue(partitionExecSend.topic(), index);
        });
        return null;
    }

}
